package com.cll.learn.rabbitmq.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.HashMap;
import java.util.Map;

/**
 * 描述信息:
 * RabbitMQ的配置类
 * 配置信息如下
 * 队列
 * 交换器
 * 绑定器
 * 客户端向rabbitmq发送消息的时候，带有routing_key和交换器
 * rabbitmq找到交换器之后，根据绑定key查询匹配routing_key的交换器或者队列，
 * banging_key指的是一种绑定关系
 * 将交换器和队列进行绑定的关系。绑定的时候需要指明使用的routing_key是什么
 * @author CLL
 * @version 1.0
 * @date 2021/2/28 10:03
 */
@Configuration
public class RabbitMQConfig {

    private static final Logger logger = LoggerFactory.getLogger(RabbitMQConfig.class);

    /**
     * 创建消息监听器的容器
     * @param connectionFactory 连接工厂
     * @return 消息监听器的容器
     */
    @Bean(name = "rabbitMessageListenerContainer")
    public DirectMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory) {
        //写的时候，默认使用DEFAULT_NUM_THREADS =Runtime.getRuntime().availableProcessors() * 2个线程
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
        // 手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 预取五个消息
        container.setPrefetchCount(5);
        // 设置每个队列供5个消费者使用
        container.setConsumersPerQueue(5);
        // 每次确认1个消息，收到一个消息就确认一次
        container.setMessagesPerAck(1);

        int availableProcessors = Runtime.getRuntime().availableProcessors();
        logger.info("===>>> 系统有效线程数 = {}", availableProcessors);

        // 设置使用的线程池
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(20);
        container.setTaskExecutor(taskExecutor);

        return container;
    }

    /**
     * 消息转换器。先转换消息再发送，先接收消息再转换
     * @return  消息转换器
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    /**
     * 订单消息队列
     * @return  订单消息队列
     */
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("queue.order").build();
    }

    /**
     * 超时消息队列
     * @return  超时消息队列
     */
    @Bean
    public Queue ttlQueue() {
        Map<String, Object> args = new HashMap<>(3);
        // 10s消息到期
        args.put("x-message-ttl", 10000);
        // 超时之后，路由到死信交换器
        args.put("x-dead-letter-exchange", "exchange.order.dlx");
        // 死信交换器的routing_key
        args.put("x-dead-letter-routing-key", "routing.order.dlx");
        return new Queue("queue.order.ttl", true, false, false, args);
    }

    /**
     * 死信队列
     * @return  死信队列
     */
    @Bean
    public Queue dlxQueue() {
        Map<String, Object> args = new HashMap<>(0);
        return new Queue("queue.order.dlx", true, false, false, args);
    }

    /**
     * 订单交换器
     * @return 订单交换器
     */
    @Bean
    public Exchange orderExchange() {
        Map<String, Object> args = new HashMap<>(0);
        return new DirectExchange("exchange.order", true, false, args);
    }

    /**
     * 超时交换器
     * @return 超时交换器
     */
    @Bean
    public Exchange ttlExchange() {
        Map<String, Object> args = new HashMap<>(0);
        return new DirectExchange("exchange.order.ttl", true, false, args);
    }

    /**
     * 死信交换器
     * @return 死信交换器
     */
    @Bean
    public Exchange dlxExchange() {
        Map<String, Object> args = new HashMap<>(0);
        return new DirectExchange("exchange.order.dlx", true, false, args);
    }

    /**
     * 用于发送下单，做分布式事务的MQ.订单队列的routing_key
     * @return  订单绑定器
     */
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("routing.order").noargs();
    }

    /**
     * 用于等待用户支付的延迟队列绑定
     * @return 支付绑定器
     */
    @Bean
    public Binding ttlBinding() {
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("routing.order.ttl").noargs();
    }

    /**
     * 用于支付超时取消用户订单的死信队列绑定
     * @return  支付超时的死信绑定器
     */
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("routing.order.dlx").noargs();
    }
}
